/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb;

import java.util.ArrayList;
import java.util.List;

import org.apache.log4j.Logger;
import org.zeromq.ZContext;
import org.zeromq.ZFrame;
import org.zeromq.ZLoop;
import org.zeromq.ZMQ; 
import org.zeromq.ZLoop.IZLoopHandler;
import org.zeromq.ZMQ.PollItem;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;
import org.zeromq.ZMsg;

public class CSBSender extends CSBServerBase implements Runnable {
	private String zkConn;
	private String frontConn;
	private String backConn;
	private int numThread;
	private int ackMode;
    private final String id = CSBUtils.sockId();
			
    private Socket frontend;
    private Socket backend;
    private List<ZFrame> workers;
    
    private static Logger LOG = Logger.getLogger(CSBSender.class.getName());
    
    public CSBSender(String zkConn, String frontConn, String backConn,
    		int numThread, int ackMode, int perfInterval) {
    	super(perfInterval);
    	this.zkConn = zkConn;
    	this.frontConn = frontConn;
    	this.backConn = backConn;
    	this.numThread= numThread;
    	this.ackMode = ackMode;
    }

    private IZLoopHandler front = new IZLoopHandler() {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
            ZMsg msg = ZMsg.recvMsg(frontend);
            if (msg != null) {            
            	ZFrame frame = workers.remove(0);
            	String s = new String(frame.getData());
            	msg.wrap(frame);
            	msg.send(backend);
            
            	mark(s);
            	if (workers.size() == 0) {
            		loop.removePoller(new PollItem(frontend, 0));
            	}
            } else {
            	LOG.error(id + ": frontend receiving interrupted.");
            }
            return 0;
        }
    };

    private IZLoopHandler back = new IZLoopHandler() {
        @Override
        public int handle(ZLoop loop, PollItem item, Object arg) {
            ZMsg msg = ZMsg.recvMsg(backend);
            if (msg != null) {            
            	ZFrame address = msg.unwrap();
            	workers.add(address);
            	
            	if (workers.size() == 1) {
            		PollItem newItem = new PollItem(frontend, ZMQ.Poller.POLLIN);
            		loop.addPoller(newItem, front, null);
            	}
            	
            	ZFrame frame = msg.peek();
            	if (new String(frame.getData()).equals(CSB.WORKER_READY))
            		msg.destroy();
            	else
            		msg.send(frontend);
            } else {
            	LOG.error(id + ": backend receiving interrupted.");
            }
        	return 0;
        }
    };
    
    public void run() {
    	perf();
    	
        workers = new ArrayList<ZFrame>();
                
    	ZContext ctx = new ZContext();
        frontend = ctx.createSocket(ZMQ.ROUTER);
        backend = ctx.createSocket(ZMQ.ROUTER);        
        frontend.bind(frontConn);          
        backend.bind(backConn);    				
        LOG.info(id + ": binding backend via " + backConn + "...");
        LOG.info(id + ": binding frontend via " + frontConn + "...");

        Sender sender = new Sender
        	(zkConn, numThread, ackMode, ctx, CSB.INPROC_CONN);
        new Thread(sender).start();
        LOG.info(id + ": sender creating...");

        ZLoop loop = new ZLoop();
    	PollItem backPoller = new PollItem(backend, Poller.POLLIN);
    	loop.addPoller(backPoller, back, null);
    	loop.start();
    	loop.destroy();
    	LOG.error(id + ": loop interrupted, exiting...");

    	while (workers.size()>0) {
            ZFrame frame = workers.remove(0);
            frame.destroy();
        }
        
        workers.clear();
        ctx.destroy();      
    }
}

